"""Copyright(c) 2023 lyuwenyu. All Rights Reserved.
"""

import copy
from collections import OrderedDict

import torch
import torch.nn as nn
import torch.nn.functional as F

from .utils import get_activation

from ...core import register

__all__ = ['HybridEncoder']


class ConvNormLayer(nn.Module):
    def __init__(self, ch_in, ch_out, kernel_size, stride, padding=None, bias=False, act=None):
        super().__init__()
        self.conv = nn.Conv2d(
            ch_in,
            ch_out,
            kernel_size,
            stride,
            padding=(kernel_size - 1) // 2 if padding is None else padding,
            bias=bias)
        self.norm = nn.BatchNorm2d(ch_out)
        self.act = nn.Identity() if act is None else get_activation(act)

    def forward(self, x):
        return self.act(self.norm(self.conv(x)))


class RepVggBlock(nn.Module):
    def __init__(self, ch_in, ch_out, act='relu'):
        super().__init__()
        self.ch_in = ch_in
        self.ch_out = ch_out
        self.conv1 = ConvNormLayer(ch_in, ch_out, 3, 1, padding=1, act=None)
        self.conv2 = ConvNormLayer(ch_in, ch_out, 1, 1, padding=0, act=None)
        self.act = nn.Identity() if act is None else get_activation(act)

    def forward(self, x):
        if hasattr(self, 'conv'):
            y = self.conv(x)
        else:
            y = self.conv1(x) + self.conv2(x)

        return self.act(y)

    def convert_to_deploy(self):
        if not hasattr(self, 'conv'):
            self.conv = nn.Conv2d(self.ch_in, self.ch_out, 3, 1, padding=1)

        kernel, bias = self.get_equivalent_kernel_bias()
        self.conv.weight.data = kernel
        self.conv.bias.data = bias

    def get_equivalent_kernel_bias(self):
        kernel3x3, bias3x3 = self._fuse_bn_tensor(self.conv1)
        kernel1x1, bias1x1 = self._fuse_bn_tensor(self.conv2)

        return kernel3x3 + self._pad_1x1_to_3x3_tensor(kernel1x1), bias3x3 + bias1x1

    def _pad_1x1_to_3x3_tensor(self, kernel1x1):
        if kernel1x1 is None:
            return 0
        else:
            return F.pad(kernel1x1, [1, 1, 1, 1])

    def _fuse_bn_tensor(self, branch: ConvNormLayer):
        if branch is None:
            return 0, 0
        kernel = branch.conv.weight
        running_mean = branch.norm.running_mean
        running_var = branch.norm.running_var
        gamma = branch.norm.weight
        beta = branch.norm.bias
        eps = branch.norm.eps
        std = (running_var + eps).sqrt()
        t = (gamma / std).reshape(-1, 1, 1, 1)
        return kernel * t, beta - running_mean * gamma / std


class CSPRepLayer(nn.Module):
    def __init__(self,
                 in_channels,
                 out_channels,
                 num_blocks=3,
                 expansion=1.0,
                 bias=None,
                 act="silu"):
        super(CSPRepLayer, self).__init__()
        hidden_channels = int(out_channels * expansion)
        self.conv1 = ConvNormLayer(in_channels, hidden_channels, 1, 1, bias=bias, act=act)
        self.conv2 = ConvNormLayer(in_channels, hidden_channels, 1, 1, bias=bias, act=act)
        self.bottlenecks = nn.Sequential(*[
            RepVggBlock(hidden_channels, hidden_channels, act=act) for _ in range(num_blocks)
        ])
        if hidden_channels != out_channels:
            self.conv3 = ConvNormLayer(hidden_channels, out_channels, 1, 1, bias=bias, act=act)
        else:
            self.conv3 = nn.Identity()

    def forward(self, x):
        x_1 = self.conv1(x)
        x_1 = self.bottlenecks(x_1)
        x_2 = self.conv2(x)
        return self.conv3(x_1 + x_2)

# transformer
class TransformerEncoderLayer(nn.Module):
    def __init__(self,
                 d_model,
                 nhead,
                 dim_feedforward=2048,
                 dropout=0.1,
                 activation="relu",
                 normalize_before=False):
        super().__init__()
        self.normalize_before = normalize_before

        self.self_attn = nn.MultiheadAttention(d_model, nhead, dropout, batch_first=True)

        self.linear1 = nn.Linear(d_model, dim_feedforward)
        self.dropout = nn.Dropout(dropout)
        self.linear2 = nn.Linear(dim_feedforward, d_model)

        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.dropout1 = nn.Dropout(dropout)
        self.dropout2 = nn.Dropout(dropout)
        self.activation = get_activation(activation)

    @staticmethod
    def with_pos_embed(tensor, pos_embed):
        return tensor if pos_embed is None else tensor + pos_embed

    def forward(self, src, src_mask=None, pos_embed=None) -> torch.Tensor:
        residual = src
        if self.normalize_before:
            src = self.norm1(src)
        q = k = self.with_pos_embed(src, pos_embed)
        src, _ = self.self_attn(q, k, value=src, attn_mask=src_mask, key_padding_mask=None)

        src = residual + self.dropout1(src)
        if not self.normalize_before:
            src = self.norm1(src)

        residual = src
        if self.normalize_before:
            src = self.norm2(src)
        src = self.linear2(self.dropout(self.activation(self.linear1(src))))
        src = residual + self.dropout2(src)
        if not self.normalize_before:
            src = self.norm2(src)
        return src


class TransformerEncoder(nn.Module):
    def __init__(self, encoder_layer, num_layers, norm=None):
        super(TransformerEncoder, self).__init__()
        self.layers = nn.ModuleList([copy.deepcopy(encoder_layer) for _ in range(num_layers)])
        self.num_layers = num_layers
        self.norm = norm

    def forward(self, src, src_mask=None, pos_embed=None) -> torch.Tensor:
        output = src
        for layer in self.layers:
            output = layer(output, src_mask=src_mask, pos_embed=pos_embed)

        if self.norm is not None:
            output = self.norm(output)

        return output


@register()
class HybridEncoder(nn.Module):
    """
       HybridEncoder是一个结合了卷积神经网络(CNN)和变换器(Transformer)的编码器模块。
       它对输入的特征进行多尺度融合和编码，输出编码后的特征图。

       参数:
       - in_channels (list): 输入特征图的通道数，默认为[512, 1024, 2048]。
       - feat_strides (list): 输入特征图的步幅，默认为[8, 16, 32]。
       - hidden_dim (int): 变换器层的隐藏维度，默认为256。
       - nhead (int): 变换器层中多头注意力机制的头数，默认为8。
       - dim_feedforward (int): 变换器层中前馈神经网络的维度，默认为1024。
       - dropout (float): 变换器层中dropout的比率，默认为0.0。
       - enc_act (str): 变换器层中的激活函数，默认为'gelu'。
       - use_encoder_idx (list): 使用编码器的索引，默认为[2]。
       - num_encoder_layers (int): 变换器编码器的层数，默认为1。
       - pe_temperature (int): 位置编码的温度参数，默认为10000。
       - expansion (float): CSPRepLayer模块的膨胀率，默认为1.0。
       - depth_mult (float): 深度乘数，用于计算CSPRepLayer的重复次数，默认为1.0。
       - act (str): 激活函数，默认为'silu'。
       - eval_spatial_size (tuple): 评估时的空间尺寸，默认为None。
       - version (str): 版本号，影响输入投影的结构，默认为'v2'。
       """
    __share__ = ['eval_spatial_size', ]

    def __init__(self,
                 in_channels=[512, 1024, 2048],
                 feat_strides=[8, 16, 32],
                 hidden_dim=256,
                 nhead=8,
                 dim_feedforward=1024,
                 dropout=0.0,
                 enc_act='gelu',
                 use_encoder_idx=[2],
                 num_encoder_layers=1,
                 pe_temperature=10000,
                 expansion=1.0,
                 depth_mult=1.0,
                 act='silu',
                 eval_spatial_size=None,
                 version='v2'):
        super().__init__()
        self.in_channels = in_channels
        self.feat_strides = feat_strides
        self.hidden_dim = hidden_dim
        self.use_encoder_idx = use_encoder_idx
        self.num_encoder_layers = num_encoder_layers
        self.pe_temperature = pe_temperature
        self.eval_spatial_size = eval_spatial_size
        self.out_channels = [hidden_dim for _ in range(len(in_channels))]
        self.out_strides = feat_strides

        # channel projection 通道投影
        self.input_proj = nn.ModuleList()
        for in_channel in in_channels:
            if version == 'v1':
                proj = nn.Sequential(
                    nn.Conv2d(in_channel, hidden_dim, kernel_size=1, bias=False),
                    nn.BatchNorm2d(hidden_dim))
            elif version == 'v2':
                proj = nn.Sequential(OrderedDict([
                    ('conv', nn.Conv2d(in_channel, hidden_dim, kernel_size=1, bias=False)),
                    ('norm', nn.BatchNorm2d(hidden_dim))
                ]))
            else:
                raise AttributeError()

            self.input_proj.append(proj)

        # encoder transformer
        encoder_layer = TransformerEncoderLayer(
            hidden_dim,
            nhead=nhead,
            dim_feedforward=dim_feedforward,
            dropout=dropout,
            activation=enc_act)

        self.encoder = nn.ModuleList([
            TransformerEncoder(copy.deepcopy(encoder_layer), num_encoder_layers) for _ in range(len(use_encoder_idx))
        ])

        # top-down fpn 自顶向下的FPN
        self.lateral_convs = nn.ModuleList()
        self.fpn_blocks = nn.ModuleList()
        for _ in range(len(in_channels) - 1, 0, -1):
            self.lateral_convs.append(ConvNormLayer(hidden_dim, hidden_dim, 1, 1, act=act))
            self.fpn_blocks.append(
                CSPRepLayer(hidden_dim * 2, hidden_dim, round(3 * depth_mult), act=act, expansion=expansion)
            )

        # bottom-up pan 自底向上的PAN
        self.downsample_convs = nn.ModuleList()
        self.pan_blocks = nn.ModuleList()
        for _ in range(len(in_channels) - 1):
            self.downsample_convs.append(
                ConvNormLayer(hidden_dim, hidden_dim, 3, 2, act=act)
            )
            self.pan_blocks.append(
                CSPRepLayer(hidden_dim * 2, hidden_dim, round(3 * depth_mult), act=act, expansion=expansion)
            )

        # 添加 BIFPN 层
        self.upbottom_conv = nn.ModuleList()
        self.bottomup_conv1 = nn.ModuleList()
        self.bottomup_conv2 = nn.ModuleList()
        for i in range(len(in_channels) - 1):
            self.upbottom_conv.append(ConvNormLayer(hidden_dim, hidden_dim, 1, 1, act=act))
            self.bottomup_conv1.append(ConvNormLayer(hidden_dim, hidden_dim, 1, 1, act=act))
            self.bottomup_conv2.append(ConvNormLayer(hidden_dim, hidden_dim, 1, 1, act=act))

        self._reset_parameters()

    def _reset_parameters(self):
        """
        初始化或重置位置嵌入参数。

        在评估模式下，根据指定的编码器索引和评估空间大小，
        为每个编码器层生成对应的位置嵌入。
        """
        if self.eval_spatial_size:
            for idx in self.use_encoder_idx:
                stride = self.feat_strides[idx]
                # 根据当前编码器层的步幅和评估空间大小生成2D正弦余弦位置嵌入
                pos_embed = self.build_2d_sincos_position_embedding(
                    self.eval_spatial_size[0] // stride, self.eval_spatial_size[1] // stride,
                    self.hidden_dim, self.pe_temperature)
                # 将生成的位置嵌入设置为模型的属性
                setattr(self, f'pos_embed{idx}', pos_embed)
                # self.register_buffer(f'pos_embed{idx}', pos_embed)

    @staticmethod
    def build_2d_sincos_position_embedding(w, h, embed_dim=256, temperature=10000.):
        """
            提供位置信息，生成二维的正弦余弦位置嵌入。

            参数:
            w: 宽度方向上的位置数量
            h: 高度方向上的位置数量
            embed_dim: 嵌入维度，默认为256
            temperature: 用于计算位置嵌入的温度参数，默认为10000

            返回:
            一个形状为(1, w*h, embed_dim)的张量，包含二维正弦余弦位置嵌入。
        """
        grid_w = torch.arange(int(w), dtype=torch.float32)
        grid_h = torch.arange(int(h), dtype=torch.float32)
        grid_w, grid_h = torch.meshgrid(grid_w, grid_h, indexing='ij')
        assert embed_dim % 4 == 0, \
            'Embed dimension must be divisible by 4 for 2D sin-cos position embedding'
        pos_dim = embed_dim // 4
        omega = torch.arange(pos_dim, dtype=torch.float32) / pos_dim
        omega = 1. / (temperature ** omega)

        out_w = grid_w.flatten()[..., None] @ omega[None]
        out_h = grid_h.flatten()[..., None] @ omega[None]
        # 按照正弦和余弦函数组合位置嵌入，并返回
        return torch.concat([out_w.sin(), out_w.cos(), out_h.sin(), out_h.cos()], dim=1)[None, :, :]

    def forward(self, feats):
        # 处理多尺度特征图   将输入特征图 feats 通过一系列线性变换投影到统一维度
        # 通过自顶向下的路径（FPN）和自底向上的路径（PAN）进行特征融合，最终输出多尺度特征图
        assert len(feats) == len(self.in_channels)
        proj_feats = [self.input_proj[i](feat) for i, feat in enumerate(feats)]

        # encoder 如果配置的有编码层则通过编码层进行特征增强
        if self.num_encoder_layers > 0:
            for i, enc_ind in enumerate(self.use_encoder_idx):
                h, w = proj_feats[enc_ind].shape[2:]
                # flatten [B, C, H, W] to [B, HxW, C]
                src_flatten = proj_feats[enc_ind].flatten(2).permute(0, 2, 1)
                if self.training or self.eval_spatial_size is None:
                    pos_embed = self.build_2d_sincos_position_embedding(
                        h, w, self.hidden_dim, self.pe_temperature).to(src_flatten.device)
                else:
                    pos_embed = getattr(self, f'pos_embed{enc_ind}', None).to(src_flatten.device)

                memory: torch.Tensor = self.encoder[i](src_flatten, pos_embed=pos_embed)
                proj_feats[enc_ind] = memory.permute(0, 2, 1).reshape(-1, self.hidden_dim, h, w).contiguous()

        # 使用 BIFPN 进行特征融合
        outs = self.bifpn(proj_feats)

        # 确保输出被正确地使用在损失计算中
        # 这里假设我们需要返回所有特征图用于损失计算
        return outs

    def _upsample_add(self, x, y):
        """Upsample and add two feature maps."""
        _, _, H, W = y.size()
        return F.interpolate(x, size=(H, W), mode='nearest') + y

    def bifpn(self, srcs):
        """
        这是EfficientDet的特征融合方式:
        paper: https://arxiv.org/abs/1911.09070
        code: https://github.com/zylo117/Yet-Another-EfficientDet-Pytorch
        """
        # reverse srcs
        feature_maps = srcs[::-1]
        up_bottom_features = []
        bottom_up_features = []
        up_bottom_features.append(feature_maps[0])
        for feature, conv in zip(feature_maps[1:], self.upbottom_conv):
            # 确保上采样后的特征图尺寸与目标特征图尺寸一致
            upsampled_feature = F.interpolate(feature, size=(feature_maps[0].size(2), feature_maps[0].size(3)), mode='nearest')
            prev_feature = self._upsample_add(up_bottom_features[0], conv(upsampled_feature))
            up_bottom_features.insert(0, prev_feature)

        bottom_up_features.append(up_bottom_features[0])
        for i in range(1, len(up_bottom_features)):
            prev_feature = self.bottomup_conv1[i - 1](bottom_up_features[0])
            # 确保所有特征图的尺寸一致
            upsampled_feature = F.interpolate(srcs[i], size=(up_bottom_features[i].size(2), up_bottom_features[i].size(3)), mode='nearest')
            prev_feature = prev_feature + self._upsample_add(up_bottom_features[i], self.bottomup_conv2[i - 1](upsampled_feature))
            bottom_up_features.insert(0, prev_feature)

        return bottom_up_features[::-1]
